package com.yeeiee.catalog.impl;

import cn.hutool.db.Entity;
import com.yeeiee.catalog.AbstractJdbcCatalog;
import com.yeeiee.config.JdbcCatalogConfig;
import com.yeeiee.convert.ClinkConverter;
import com.yeeiee.convert.impl.MysqlConverter;
import lombok.NoArgsConstructor;
import lombok.NonNull;
import lombok.extern.slf4j.Slf4j;
import lombok.val;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.table.api.Schema;
import org.apache.flink.table.catalog.*;
import org.apache.flink.table.catalog.exceptions.*;
import org.apache.flink.util.Preconditions;

import javax.annotation.Nullable;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.regex.Pattern;
import java.util.stream.Collectors;

import static org.apache.flink.connector.jdbc.table.JdbcConnectorOptions.*;
import static org.apache.flink.table.factories.FactoryUtil.CONNECTOR;


@Slf4j
@NoArgsConstructor
public class MysqlCatalog extends AbstractJdbcCatalog {

    private ClinkConverter clinkConverter;
    private String connector;
    private String suffix;

    public MysqlCatalog(
            ClassLoader classLoader,
            @NonNull String url,
            @NonNull String defaultDatabase,
            @Nullable String driverName,
            @Nullable String defaultSchema,
            @Nullable String username,
            @Nullable String password,
            ClinkConverter clinkConverter,
            String suffix) {
        super(classLoader, url, defaultDatabase, driverName, defaultSchema, username, password);
        this.clinkConverter = clinkConverter;
        this.suffix = suffix;
    }

    @Override
    protected Pattern urlPattern() {
        return Pattern.compile(
                "^(?<url>jdbc:.+?//(?<host>.+?):(?<port>\\d+?))(/(?<database>.*?))*(?<suffix>\\?.*)*$");
    }

    @Override
    public void setConnector(String connector) {
        this.connector = connector;
    }

    @Override
    public String quoteIdentifier(String tableName) {
        return "`" + tableName + "`";
    }

    @Override
    public String factoryIdentifier() {
        return "mysql";
    }

    @Override
    public AbstractJdbcCatalog build(Configuration configuration) {
        val url = configuration.get(JdbcCatalogConfig.url);
        val driverName = configuration.get(JdbcCatalogConfig.driver);
        val username = configuration.getString(JdbcCatalogConfig.username, "");
        val password = configuration.getString(JdbcCatalogConfig.password, "");

        val matcher = urlPattern().matcher(url);
        if (!matcher.find()) {
            throw new CatalogException("failed url pattern: " + url);
        }
        val database = matcher.group("database");
        if (database == null || database.trim().length() == 0) {
            throw new CatalogException("failed url pattern: " + url);
        }
        val suffix = matcher.group("suffix");

        return new MysqlCatalog(
                this.getClass().getClassLoader(),
                url,
                database,
                driverName,
                null,
                username,
                password,
                new MysqlConverter(),
                suffix);
    }

    @Override
    public List<String> listDatabases() throws CatalogException {
        val entityList = queryList("SELECT `SCHEMA_NAME` FROM `INFORMATION_SCHEMA`.`SCHEMATA`");
        return entityList.stream().map(e -> e.getStr("SCHEMA_NAME")).collect(Collectors.toList());
    }

    @Override
    public void createDatabase(String name, CatalogDatabase database, boolean ignoreIfExists) throws DatabaseAlreadyExistException, CatalogException {
        Preconditions.checkNotNull(name, "database name cannot be null");
        val databaseExists = super.databaseExists(name);
        if (databaseExists && !ignoreIfExists) {
            throw new DatabaseAlreadyExistException(this.factoryIdentifier(), name);
        }
        execute("CREATE DATABASE IF NOT EXISTS" + quoteIdentifier(name));
    }

    @Override
    public void dropDatabase(String name, boolean ignoreIfNotExists, boolean cascade) throws DatabaseNotExistException, DatabaseNotEmptyException, CatalogException {
        Preconditions.checkNotNull(name, "database name cannot be null");
        val databaseExists = super.databaseExists(name);
        if (!databaseExists && !ignoreIfNotExists) {
            throw new DatabaseNotExistException(this.factoryIdentifier(), name);
        }
        execute("DROP DATABASE IF EXISTS " + quoteIdentifier(name));
    }

    @Override
    public List<String> listTables(String databaseName) throws DatabaseNotExistException, CatalogException {
        Preconditions.checkNotNull(databaseName, "Database name cannot be null");
        if (!super.databaseExists(databaseName)) {
            throw new DatabaseNotExistException(this.factoryIdentifier(), databaseName);
        }
        val entityList = queryList("SELECT TABLE_NAME FROM information_schema.`TABLES` WHERE TABLE_SCHEMA = ?", databaseName);
        return entityList.stream().map(e -> e.getStr("TABLE_NAME")).collect(Collectors.toList());
    }

    @Override
    public CatalogBaseTable getTable(ObjectPath tablePath) throws TableNotExistException, CatalogException {
        Preconditions.checkNotNull(tablePath, "Table path cannot be null");
        // 表不存在抛出异常
        if (!super.tableExists(tablePath)) {
            throw new TableNotExistException(this.factoryIdentifier(), tablePath);
        }
        val databaseName = tablePath.getDatabaseName();
        val tableName = tablePath.getObjectName();

        val primaryKey = getPrimaryKey(databaseName, null, tableName);
        val entityList = queryList("select\n" +
                "    COLUMN_NAME as `column_name`,\n" +
                "    COLUMN_TYPE as `source_type`,\n" +
                "    DATA_TYPE as `type_name`,\n" +
                "    NUMERIC_PRECISION as `precision`,\n" +
                "    NUMERIC_SCALE as `scale`,\n" +
                "    if(UPPER(IS_NULLABLE)='YES',1,0) as `is_nullable`,\n" +
                "    CHARACTER_MAXIMUM_LENGTH as `column_character_length`,\n" +
                "    COLUMN_COMMENT as `comment`\n" +
                "from INFORMATION_SCHEMA.COLUMNS\n" +
                "where TABLE_SCHEMA = ?\n" +
                "and TABLE_NAME = ?\n" +
                "order by ORDINAL_POSITION;", databaseName, tableName);

        // column
        val builder = Schema.newBuilder();
        for (Entity entity : entityList) {
            clinkConverter.convertToType(entity, builder);
        }

        // pk
        primaryKey.map(UniqueConstraint::getColumns)
                .ifPresent(builder::primaryKey);

        // option
        Map<String, String> props = new HashMap<>();
        props.put(CONNECTOR.key(), this.connector);
        props.put(URL.key(), super.getUrl());
        props.put(USERNAME.key(), super.getUsername());
        props.put(PASSWORD.key(), super.getPassword());
        props.put(TABLE_NAME.key(), tablePath.getObjectName());

        // table comment
        val entity = queryOne("select\n" +
                "\t`TABLE_COMMENT` as table_comment\n" +
                "from information_schema.tables \n" +
                "where TABLE_SCHEMA = ? and TABLE_NAME = ?", databaseName, tableName);
        val tableComment = entity.getStr("table_comment");

        val catalogTable = CatalogTable.of(
                builder.build(),
                tableComment,
                Collections.emptyList(),
                props
        );

        log.info("Created catalog table: {}", catalogTable);
        return catalogTable;
    }

    @Override
    public void dropTable(ObjectPath tablePath, boolean ignoreIfNotExists) throws TableNotExistException, CatalogException {
        Preconditions.checkNotNull(tablePath, "tablePath cannot be null");
        if (!super.tableExists(tablePath)) {
            throw new TableNotExistException(this.factoryIdentifier(), tablePath);
        }
        execute("DROP TABLE IF EXISTS " + quoteIdentifier(tablePath));
    }

    @Override
    public void createTable(ObjectPath tablePath, CatalogBaseTable table, boolean ignoreIfExists) throws TableAlreadyExistException, DatabaseNotExistException, CatalogException {
        Preconditions.checkNotNull(tablePath, "tablePath cannot be null");
        val databaseName = tablePath.getDatabaseName();
        val tableName = tablePath.getObjectName();
        if (!super.databaseExists(databaseName)) {
            throw new DatabaseNotExistException(this.factoryIdentifier(), databaseName);
        }

        val tableExists = super.tableExists(tablePath);
        if (tableExists && !ignoreIfExists) {
            throw new TableAlreadyExistException(this.factoryIdentifier(), tablePath);
        }

        // column
        val builder = new StringBuilder();
        builder.append("CREATE TABLE IF NOT EXISTS ")
                .append(quoteIdentifier(databaseName, tableName))
                .append(" (");
        val unresolvedSchema = table.getUnresolvedSchema();
        val columns = unresolvedSchema.getColumns();
        for (Schema.UnresolvedColumn column : columns) {
            val typeString = clinkConverter.convertToDb(column);
            val name = column.getName();
            val comment = column.getComment().orElse("");

            builder.append(quoteIdentifier(name))
                    .append(" ")
                    .append(typeString)
                    .append(" COMMENT '")
                    .append(comment)
                    .append("',");
        }

        // primary key
        val primaryKey = unresolvedSchema.getPrimaryKey();
        if (primaryKey.isPresent()) {
            val joinedColumnNames = primaryKey.get().getColumnNames().stream().map(this::quoteIdentifier).collect(Collectors.joining(","));
            builder.append("PRIMARY KEY (")
                    .append(joinedColumnNames)
                    .append(")");
        } else {
            builder.deleteCharAt(builder.length() - 1);
        }

        // table comment
        val tableComment = table.getComment();

        builder.append(") COMMENT '").append(tableComment).append("'");

        // create table
        execute(builder.toString());
    }
}
